package org.raymond.iworks.study.mq.kafka.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.TopicConfig;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class AdminTest {
    private static final String kafkaAddress = "192.168.0.125:9094,192.168.0.125:9194,192.168.0.125:9294";
    public static void main(String[] args) throws Exception {
        String topicName = "my-topic";
        //deleteTopic(topicName);
        createTopic(topicName);
        queryTopic();
    }

    public static void createTopic(String topicName) throws Exception {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);

        try (Admin admin = Admin.create(props)) {
            int partitions = 3;
            short replicationFactor = 2;
            // Create a compacted topic
            log.info("try create a topic");
            CreateTopicsResult result = admin.createTopics(Collections.singleton(
                    new NewTopic(topicName, partitions, replicationFactor)
                            .configs(Collections.singletonMap(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT))));

            // Call values() to get the result for a specific topic
            KafkaFuture<Void> future = result.values().get(topicName);

            // Call get() to block until the topic creation is complete or has failed
            // if creation failed the ExecutionException wraps the underlying cause.
            future.get();
            log.info("created a topic finish");
        }
    }

    public static void queryTopic() throws Exception {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);

        try (Admin admin = Admin.create(props)) {
            log.info("try list");
            ListTopicsResult listTopicsResult = admin.listTopics();
            Collection<TopicListing> topicListings = listTopicsResult.listings().get();
            topicListings.forEach(c->{
                log.info("topic:{}, topicId:{}", c.toString(), c.topicId().toString());
            });

            Collection<ConsumerGroupListing> consumerGroupListings = admin.listConsumerGroups().all().get();
            consumerGroupListings.forEach(c->{
                log.info("consumerGroup:{}", c.toString());

                try {
                    Map<String, Map<TopicPartition, OffsetAndMetadata>> consumerGroupOffsets = admin.listConsumerGroupOffsets(c.groupId()).all().get();
                    consumerGroupOffsets.forEach((partition, map)->{
                        OffsetAndMetadata oam = map.get(partition);
                        log.info("partition: {}, offsetAndMeta:{}", partition, oam);
                    });
                } catch (InterruptedException | ExecutionException e) {
                    log.error("{}", e.getMessage());
                }
            });

            log.info("list finish");
        }
    }

    public static void deleteTopic(String topic){
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);

        try (Admin admin = Admin.create(props)) {
            admin.deleteTopics(List.of(topic));
        }
    }
}
